package beautiful.butterfly.drds.data_exchange.core.transport.exchanger;

import beautiful.butterfly.drds.data_exchange.core.statistics.communication.Communication;
import beautiful.butterfly.drds.data_exchange.core.statistics.communication.CommunicationTool;
import beautiful.butterfly.drds.data_exchange.core.transport.transformer.TransformerErrorCode;
import beautiful.butterfly.drds.data_exchange.core.transport.transformer.TransformerExecution;
import common.element.Record;
import common.exception.DataExchangeException;
import common.plugin.TaskPluginCollector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;


public abstract class TransformerExchanger
{

    private static final Logger LOG = LoggerFactory.getLogger(TransformerExchanger.class);
    protected final TaskPluginCollector pluginCollector;

    protected final int taskGroupId;
    protected final int taskId;
    protected final Communication communication;

    private long totalExaustedTime = 0;
    private long totalFilterRecords = 0;
    private long totalSuccessRecords = 0;
    private long totalFailedRecords = 0;


    private List<TransformerExecution> transformerExecutionList;


    public TransformerExchanger(int taskGroupId, int taskId, Communication communication,
                                List<TransformerExecution> transformerExecutionList,
                                final TaskPluginCollector pluginCollector)
    {

        this.transformerExecutionList = transformerExecutionList;
        this.pluginCollector = pluginCollector;
        this.taskGroupId = taskGroupId;
        this.taskId = taskId;
        this.communication = communication;
    }


    public Record doTransformer(Record record)
    {
        if (transformerExecutionList == null || transformerExecutionList.size() == 0)
        {
            return record;
        }

        Record newRecord = record;

        long diffExaustedTime = 0;
        String errorMsg = null;
        boolean failed = false;
        for (TransformerExecution transformerExecution : transformerExecutionList)
        {
            long startTs = System.nanoTime();


            /**
             * 延迟检查transformer参数的有效性，直接抛出异常，不作为脏数据
             * 不需要在插件中检查参数的有效性。但参数的个数等和插件相关的参数，在插件内部检查
             */
            if (!transformerExecution.isChecked())
            {

                if (transformerExecution.getColumnIndex() != null && transformerExecution.getColumnIndex() >= record.getColumnListSize())
                {
                    throw DataExchangeException.asDataExchangeException(TransformerErrorCode.TRANSFORMER_ILLEGAL_PARAMETER,
                            String.format("columnIndex[%s] out of bound[%s]. name=%s",
                                    transformerExecution.getColumnIndex(), record.getColumnListSize(),
                                    transformerExecution.getTransformerName()));
                }
                transformerExecution.setIsChecked(true);
            }

            try
            {
                newRecord = transformerExecution.getTransformer().evaluate(newRecord, transformerExecution.gettContext(), transformerExecution.getFinalParas());
            } catch (Exception e)
            {
                errorMsg = String.format("transformer(%s) has Exception(%s)", transformerExecution.getTransformerName(),
                        e.getMessage());
                failed = true;
                break;

            } finally
            {

            }

            if (newRecord == null)
            {
                /**
                 * 这个null不能传到writer，必须消化掉
                 */
                totalFilterRecords++;
                //transformerInfoExec.addFilterRecords(1);
                break;
            }

            long diff = System.nanoTime() - startTs;
            //transformerInfoExec.addExaustedTime(diff);
            diffExaustedTime += diff;
            //transformerInfoExec.addSuccessRecords(1);
        }

        totalExaustedTime += diffExaustedTime;

        if (failed)
        {
            totalFailedRecords++;
            this.pluginCollector.collectDirtyRecord(record, errorMsg);
            return null;
        } else
        {
            totalSuccessRecords++;
            return newRecord;
        }
    }

    public void doStat()
    {

        /**
         * todo 对于多个transformer时，各个transformer的单独统计进行显示。最后再汇总整个transformer的时间消耗.
         * 暂时不统计。
         */
        communication.setLongCounter(CommunicationTool.transformer_succeed_records, totalSuccessRecords);
        communication.setLongCounter(CommunicationTool.transformer_failed_records, totalFailedRecords);
        communication.setLongCounter(CommunicationTool.transformer_filter_records, totalFilterRecords);
        communication.setLongCounter(CommunicationTool.transformer_used_time, totalExaustedTime);
    }


}
